package spark.sql

import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}

/**
  *
  * @author com.ymy.hadoop   
  * @since 2019/6/4 22:33
  */

// spark 常见操作
object SparkDemo03 {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("SparkDemo03")
      .master("local[*]")
      .getOrCreate()

    val schema = StructType(Array(
      StructField("name",StringType,false),
      StructField("age",IntegerType,false),
      StructField("address",StringType,false),
      StructField("birthday",DateType,false)
    ))

    val rowRDD = spark.sparkContext.parallelize(Seq(
      Row("zhangsan",10,"beijing",java.sql.Date.valueOf("2008-01-01")),
      Row("lisi",20,"shanghai",java.sql.Date.valueOf("1998-01-01")),
      Row("wangwu",70,"shanghai",java.sql.Date.valueOf("1998-01-01")),
      Row("zhaoliu",10,"shanghai",java.sql.Date.valueOf("1998-01-01"))
    ), 2)

    val df = spark.createDataFrame(rowRDD,schema)

    import org.apache.spark.sql.functions._

    // 重要：后面的select('*) 会自动识别，哪些列可以出现在group by之后
//    df.groupBy('address).agg(("age","sum")).select('*).show(false)
    df.groupBy("address").agg(Map("age"->"avg")).select(expr("*")).show(false)

//    df.select(col = "name").show(false)
//    df.select(col("name")).show(false)
//    df.select("name").show(false)
//    df.select('name,'age).show(false)
//    df.select($"name").show(false)

//    df.select(
//      df.col("name"),
//      col("name"),
//      column("name"),
//      'name,
//      $"name",
//      expr("name"))
//      .show(2)

    // 下面两种select类型不能混用
//    df.select(
//      "name",expr("name")
//    )
//      .show(2)

    // 下面情况可以混用
//    df.select(
//      'name,col("name"),expr("name")
//    )
//      .show(2)

    // 给列命名别名
//    df.select('name.as("xxx")).show(false)

    // 添加一列叫one，并命名
//    df.select('*,lit(1).as("one")).show(false)

    // 过滤行
//    df.filter(expr("age > 15")).show(false)

    // 添加列
//    df.withColumn("add",lit(1)).show(false)

//    df.withColumn("witgFilter",expr("name=='lisi'")).show(false)
//    +--------+---+--------+----------+----------+
//    |name    |age|address |birthday  |witgFilter|
//    +--------+---+--------+----------+----------+
//    |zhangsan|10 |beijing |2008-01-01|false     |
//      |lisi    |20 |shanghai|1998-01-01|true      |
//      +--------+---+--------+----------+----------+

    // 打印出所有的列
//    println(df.withColumn("name",expr("age")).columns.mkString(","))
//    println(df.columns.mkString(","))

    // 重命名列
//     df.withColumnRenamed("name","newName").columns.foreach(println)

    // 删除列
//    df.drop("name").show(false)

    //  等价SQL：select *,cast(age as string) as xxoo from dfTable;
//    df.withColumn("xxoo",expr("age").cast("string")).show(false)

    spark.stop()
  }
}
